package com.ruoyi.system.mqtt;

import io.moquette.broker.Server;
import io.moquette.broker.config.ClasspathResourceLoader;
import io.moquette.broker.config.IConfig;
import io.moquette.broker.config.ResourceLoaderConfig;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.io.IOException;

/**
 * 创建mqtt broker服务器
 */
@Slf4j
@Service
public class MqttBroker {

//    static {
//        try {
//            init();
//        } catch (IOException e) {
//            throw new RuntimeException(e);
//        } catch (MqttException e) {
//            throw new RuntimeException(e);
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
//    }
//
//    public static void init() throws IOException, MqttException, InterruptedException {
//        // 加载配置文件（默认从classpath读取moquette.conf）
//        IConfig config = new ResourceLoaderConfig(new ClasspathResourceLoader());
//
//        // 初始化Broker
//        Server mqttBroker = new Server();
//        mqttBroker.startServer(config);
//
//        // 添加关闭钩子
//        Runtime.getRuntime().addShutdownHook(new Thread(mqttBroker::stopServer));
//        log.info("MQTT Broker启动成功，监听端口: " + config.getProperty("port"));
//
//
//        // 创建客户端
////        createClient();
//    }

    private void createClient() throws MqttException, InterruptedException {
        String brokerUrl = "tcp://172.16.9.7:1883";
        String clientId = "test";

        // 创建客户端
        MqttClient client = new MqttClient(brokerUrl, clientId);
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);

        // 3. 设置回调（异步接收消息）
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                log.info("连接断开: " + cause.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                log.info("收到消息: ");
                log.info("主题: " + topic);
                log.info("内容: " + new String(message.getPayload()));
                log.info("QoS: " + message.getQos());
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                // 发布消息完成的回调（订阅者无需处理）
            }
        });

        // 4. 连接Broker
        client.connect(options);
        log.info("已连接到Broker");
        // 5. 订阅主题
        client.subscribe("test/topic", 1);
        log.info("已订阅主题: " + "test/topic");
// 发布消息
//        client.publish("/topic", new MqttMessage("Hello MQTT!".getBytes()));
        // 6. 保持程序运行（避免退出）
//        while (true) {
//            Thread.sleep(1000);
//
//
//            // 发布消息
//            client.publish("/topic", new MqttMessage("Hello MQTT!".getBytes()));
////            client.disconnect();
//        }
    }
}